一段客服紀錄要交給摘要服務，但姓名、電話與地址不能跟著送出去。遮掉個資之後，還得留下「商品尚未送達，需要查詢配送進度」這件事。
前幾天分別談了資料怎麼準備、怎麼檢查，以及判斷有疑問時由誰接手。今天沿用同一筆案例，把這些方法放回處理過程，再用簡單模擬確認資料最後怎麼送出。
沿用 Day 11 的 PII-02，原文是固定的虛構資料：
顧客林小安來電，電話0912-000-000，地址臺中市西屯區範例街8號；反映商品尚未送達，請客服查詢配送進度。
這筆案例設定下游可以收到遮罩文字，不能取得三個個資原值，所以預期處置是 MASK。要驗收的結果，是下游收到遮罩後的內容，而且仍看得出商品未送達與配送查詢任務。
這承接 Day 5 的做法：把政策寫成可核對的行為。只寫「有個資要處理」，工程師還得猜要阻擋、遮罩，還是依授權放行。先固定這筆案例的要求，後面才有比較依據。
這次先固定虛構姓名、電話、地址，以及客服要完成的任務，也就是前文的種子資料（Seed）。準備原文、遮罩版本與個資位置時，各階段（Stage）分開責任，每份產物都用 source_ref 記下所依據的原文。
姓名、電話格式或說話方式可以在其他測試版本中改變，但下游權限與預期處置要由案例設計固定。不能因為生成文字寫了「這個服務可以取得電話」，就替它補上授權。這輪只使用事先設定的摘要路徑，避免一邊測試、一邊改變原本要求。
這就是依賴關係（Dependency）在案例中的用途。若改了原文，卻拿舊電話、舊位置或另一份遮罩來驗收，單看每份內容都可能合理，合起來卻是在比較不同資料。
品質檢查也放在這裡：必要欄位是否存在、標註位置是否切回原值、遮罩後是否殘留已知個資。保留原文與中間結果，失敗時才知道哪一步需要重做。這一段直接沿用 Day 11 的最小管線，不必先生成一大批資料。
為了說明判斷，準備三份示意遮罩結果：
0912-000-000 卻還在。依 Day 12 的規則檢查，A 可以直接比對已知電話，指出殘留原值，退回遮罩階段。這個錯誤不需要先排名，也不必請模型判斷文字是否流暢。
B 與 C 都沒有三個已知原值，規則會通過，但 B 已經讓客服任務變模糊。這正是 Day 13 的模型評審（Judge） 要處理的問題：對照原文與候選，依準則核對任務資訊。這組固定句可以直接檢查指定段落；若候選換了說法，或內容較複雜，就要保留語意判斷的空間，不能把所有字句不同都當成錯誤。
例如「商品尚未送達」改成「尚未收到商品」，在本例仍能表達同一個問題。判斷時要分別核對商品未送達、需要查詢配送這兩項資訊，並指出哪段候選文字支持判斷。只回一句「內容很好」，無法知道檢查過什麼。
如果同一筆原文有很多候選，Day 15 的獎勵模型（Reward Model） 可以協助安排先審哪些。排名高仍不能抵銷個資殘留或任務缺漏。Day 14 也提醒評估器可能判錯：若理由和原文對不起來，就要回頭核對；準則說不清楚時，才依 Day 16 交給人處理。
Day 17 串起的分工，在這裡是用來定位問題，並非每份資料都必須走完四關。本文沒有新增 Reward、Judge 或人工審查結果；程式中 B、C 的語意檢查仍是 NOT_RUN。C 是本例的參考文字，不因規則通過就變成整批資料已核准。
import hashlib
import json
import re
# Seed 固定任務與虛構個資；預期處置由案例設計決定。
VALUES = {"NAME": "林小安", "PHONE": "0912-000-000", "ADDRESS": "臺中市西屯區範例街8號"}
TASK = "反映商品尚未送達，請客服查詢配送進度。"
SOURCE = f"顧客{VALUES['NAME']}來電，電話{VALUES['PHONE']}，地址{VALUES['ADDRESS']}；{TASK}"
EXPECTED = f"顧客[NAME]來電，電話[PHONE]，地址[ADDRESS]；{TASK}"
# 此偵測器只處理本例句型；它不使用 VALUES 或 EXPECTED 當作搜尋答案。
PATTERNS = (
("NAME", re.compile(r"顧客(?P<value>[\u4e00-\u9fff]{2,4})來電")),
("PHONE", re.compile(r"(?<![0-9])(?P<value>09(?:[0-9]{8}|[0-9]{2}-[0-9]{3}-[0-9]{3}))(?![0-9])")),
("ADDRESS", re.compile(r"地址(?P<value>[\u4e00-\u9fff][^；;\n]*)(?=[；;]|$)")),
)
def require(condition, message):
"""驗收不符就報錯，避免程式執行完卻默默略過異常。"""
if not condition:
raise AssertionError(message)
def prepare():
"""原文、標註與三份示意候選都指向同一個 source_ref。"""
source_ref = "PII-02/" + hashlib.sha256(SOURCE.encode("utf-8")).hexdigest()
entities = []
for kind, value in VALUES.items():
start = SOURCE.index(value)
entities.append({"type": kind, "text": value, "start": start,
"end": start + len(value), "source_ref": source_ref})
texts = {
"A": EXPECTED.replace("[PHONE]", VALUES["PHONE"]), # 故意留下電話。
"B": EXPECTED.partition("；")[0] + "；顧客反映商品問題，請客服協助處理。", # 任務資訊變模糊。
"C": EXPECTED, # 本例預期文字，不代表已通過模型或人工審查。
}
candidates = {key: {"text": text, "source_ref": source_ref} for key, text in texts.items()}
return source_ref, entities, candidates
def check_candidate(candidate, source_ref):
"""Rule 只核對結構、來源版本及已知原值；不替代語意判斷。"""
if not isinstance(candidate, dict) or set(candidate) != {"text", "source_ref"}:
return ["SCHEMA_INVALID"]
if not all(isinstance(candidate[key], str) for key in ("text", "source_ref")):
return ["SCHEMA_INVALID"]
reasons = []
if candidate["source_ref"] != source_ref:
reasons.append("SOURCE_REF_MISMATCH")
if any(value in candidate["text"] for value in VALUES.values()):
reasons.append("KNOWN_PII_REMAINS")
return reasons
def guard(text):
"""本例固定允許送出遮罩文字；以窄範圍 Regex 找位置再替換。"""
matches = sorted((m.start("value"), m.end("value"), kind)
for kind, pattern in PATTERNS for m in pattern.finditer(text))
masked = text
# 從尾端開始，前方索引才不會因替換字數不同而位移。
for start, end, kind in reversed(matches):
masked = masked[:start] + f"[{kind}]" + masked[end:]
return ("MASK" if matches else "ALLOW"), masked
class MockReceiver:
"""下游模擬函式只記錄收到的內容，不生成摘要、不連線。"""
def __init__(self):
self.receipts = []
def receive(self, text):
self.receipts.append(text)
def run(forward_original):
"""先完成防護判斷，再模擬傳送；兩版只改選用哪份文字。"""
action, approved = guard(SOURCE)
receiver = MockReceiver()
outgoing = SOURCE if forward_original else approved
receiver.receive(outgoing)
# 結果以模擬接收端記錄為準，而不是傳送端的成功訊息。
received = receiver.receipts[0]
checks = {
"policy": action == "MASK" and approved == EXPECTED,
"calls": len(receiver.receipts) == 1,
"received_text": received == EXPECTED,
"no_known_pii": all(value not in received for value in VALUES.values()),
"task_preserved": received.partition("；")[2] == TASK,
}
return {"action": action, "calls": len(receiver.receipts), "received": received, "checks": checks,
"result": "PASS" if all(checks.values()) else "FAIL"}
def main():
source_ref, entities, candidates = prepare()
# 先核對資料標註，確認可重跑；此處沒有執行語意模型。
require(prepare() == (source_ref, entities, candidates), "相同輸入的資料不一致")
require(all(SOURCE[e["start"]:e["end"]] == e["text"] for e in entities), "標註位置錯誤")
for key, candidate in candidates.items():
reasons = check_candidate(candidate, source_ref)
require(reasons == (["KNOWN_PII_REMAINS"] if key == "A" else []), "候選檢查與預期不同")
print(f'{key}: rule={"FAIL" if reasons else "PASS"}, semantic=NOT_RUN')
old, fixed = run(forward_original=True), run(forward_original=False)
# v0 必須只在收件全文與原值殘留失敗；v1 必須全部通過。
require({key for key, ok in old["checks"].items() if not ok} ==
{"received_text", "no_known_pii"}, "故障不是原先設計的問題")
require(all(fixed["checks"].values()), "修正版未通過驗收")
for version, result in (("v0", old), ("v1", fixed)):
print(f'{version}: action={result["action"]}, calls={result["calls"]}, result={result["result"]}')
print("v1 收件：", fixed["received"])
if __name__ == "__main__":
main()
本次實際執行輸出：
A: rule=FAIL, semantic=NOT_RUN
B: rule=PASS, semantic=NOT_RUN
C: rule=PASS, semantic=NOT_RUN
v0: action=MASK, calls=1, result=FAIL
v1: action=MASK, calls=1, result=PASS
v1 收件： 顧客[NAME]來電，電話[PHONE]，地址[ADDRESS]；反映商品尚未送達，請客服查詢配送進度。
電話殘留，先修遮罩；配送資訊遺漏，檢查改寫與評估條件；政策提供正確文字卻仍送出原文，就修組成請求的程式。把這些錯誤混成一個低分，很容易改了不相關的地方，然後重新生成整批資料。
因此重跑時保留同一筆案例、原文與預期，只改出錯的部分。若改了遮罩，就重新核對個資與任務；若改了傳送，就再看接收端的紀錄。原文或候選變更後，也不能沿用舊版本的評估結果。
這筆客服紀錄最後要做到的事很清楚：摘要服務看得到配送問題，看不到顧客個資。用同一筆資料串起前幾天的方法，才看得出每個檢查在回答什麼問題，哪些結果還不足以下判斷。
如果要用在自己的流程，可以先挑一筆資料，寫下允許送出的內容，再故意讓其中一步出錯。檢查能不能指出原因，修正後用同一筆資料重跑。先把這筆資料的成敗說清楚，再增加案例，才能知道新資料補上了哪些尚未測到的情況。
iThome鐵人賽
看影片追技術